package top.atcra.rpc.protection.health;

import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import top.atcra.rpc.RPCBootstrap;
import top.atcra.rpc.config.ConfigLoader;
import top.atcra.rpc.protection.loadBalancer.LoadBalancer;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;

/**
 * 服务上下线监听器
 *
 * @author atcra
 * @date 2025-3-28
 */
@Slf4j
public class ServiceUpAndDownWatcher implements Watcher {
    protected String serviceName;
    protected String group;
    protected final RPCBootstrap instance;
    protected final Map<InetSocketAddress, Channel> channelCache;

    public ServiceUpAndDownWatcher(RPCBootstrap instance) {
        this.instance = instance;
        channelCache = instance.getChannelCache();
    }

    @Override
    public void process(WatchedEvent event) {
        // 1. 检查事件类型
        if (event.getType() != Event.EventType.NodeDeleted) {
            return;
        }
        getServiceNameAndGroup(event.getPath());
        if (log.isDebugEnabled()) {
            log.debug("检测到服务【{}[{}]】下有节点上/下线，将重新拉取服务列表...", serviceName, group);
        }
        List<InetSocketAddress> addresses = instance.getServiceRegistry()
                .lookup(serviceName, ConfigLoader.getConfig().getGroupName());
        // 处理新增的节点(在address 不在CHANNEL_CACHE)
        for (InetSocketAddress address : addresses) {
            if (!channelCache.containsKey(address)) {
                // 根据地址建立连接，并且缓存
                try {
                    Channel channel = instance.createChannel(address);
                    channelCache.put(address, channel);
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }
        }

        // 处理下线的节点(在CHANNEL_CACHE 不在address)
        channelCache.forEach(
                (inetSocketAddress, channel) -> {
                    if (!addresses.contains(inetSocketAddress)) {
                        channelCache.remove(inetSocketAddress);
                    }
                }
        );

        // 获得负载均衡器，进行重新的loadBalance

        LoadBalancer loadBalancer = instance.getLoadBalancer();
        loadBalancer.reBalance(serviceName, group, addresses);
        // ServiceHealth 重新检测
        instance.getServiceHealthDetector().checkHealth(serviceName, group);
    }

    private void getServiceNameAndGroup(String path) {
        int pos1 = path.lastIndexOf("/");
        int pos2 = path.lastIndexOf("/", pos1 - 1);
        int pos3 = path.lastIndexOf("/", pos2 - 1);
        serviceName = path.substring(pos3 + 1, pos2);
        group = path.substring(pos2 + 1, pos1);
    }
}
